//==========================================================================
// Copyright (c) 2000-2008,  Elastos, Inc.  All Rights Reserved.
//==========================================================================

#include <compoundChannel.h>
#include <osdepend.h>
#include "ChannelCommand.h"
#include "PkgManager.h"
#include <CommuChannel.h>

#ifdef _win32
#define INFINITE      0xFFFFFFFF
#endif

#define MAX_CHANNEL_NUMBER 0xFFFF

#define ROUNDSIZE(data_size) ((data_size&0x3) ? ((data_size&~0x3)+0x4) : data_size)

C_PVOID CommunicationObject::m_sThreadHandle = NULL;
C_UINT  CommunicationObject::m_sGenerateChannelNumber = 0;
C_LOCK  CommunicationObject::m_sGenerateChannelNumberLock = NULL;
list CommunicationObject::m_sInitList;
C_LOCK CommunicationObject::m_sInitListLock = NULL;
list CommunicationObject::m_sRunningList;
C_LOCK CommunicationObject::m_sRunnListLock = NULL;

C_INT CommunicationObject::Read(C_VOID *pBuffer, C_UINT nToRead, C_UINT *nRead)
{
    C_INT ret = 0;
    if ((NULL == pBuffer) || (0 == nToRead) || (NULL == nRead)) {
        return C_INVALID_ARGUMENT;
    }
    if (stopping == m_status) {
        return C_CLOSED;
    }
    *nRead = 0;

    Lock(m_lock);

//    assert(0 == m_receiveSize);
//    assert(NULL == m_pReceiveAddr);

    //Read data
    //1.Storage has data, and large enough.
    if (m_receiveStorageSize > nToRead) {
        memcpy(pBuffer, m_pReceiveStorageAddr, nToRead);
        *nRead = nToRead;
        C_UINT size = m_receiveStorageSize - nToRead; // > 0
        C_VOID *pTemp = MALLOC(size);
		if (NULL == pTemp) {
			COMMU_PRINTF("Err: Malloc failed! file%s. line:%d.\n",
					__FILE__, __LINE__);
			UnLock(m_lock);
			return C_OUT_OF_MEMORY;
		}
		memcpy(pTemp, m_pReceiveStorageAddr + nToRead, size);
		FREE(m_pReceiveStorageAddr);
		m_pReceiveStorageAddr = (C_CHAR *) pTemp;
		m_receiveStorageSize = size;
        UnLock(m_lock);
    }
    else if (m_receiveStorageSize != 0) {
        memcpy(pBuffer, m_pReceiveStorageAddr, m_receiveStorageSize);
        *nRead = m_receiveStorageSize;
        FREE(m_pReceiveStorageAddr);
        m_pReceiveStorageAddr = NULL;
        m_receiveStorageSize = 0;//Set the size to signal no data in storage.
        UnLock(m_lock);
    }
    else {
        /*
         * Allocate a block of new memory, so switching processes in kernel does
         * not affect reading from compound channel
         */

        m_receiveSize = nToRead;
        m_receiveOffset = 0;
        m_pReceiveAddr = (C_CHAR *)MALLOC(ROUNDSIZE(m_receiveSize));
        if (m_pReceiveAddr == NULL) {
            COMMU_PRINTF("Err: Out of memory file%s. line:%d.\n",
                    __FILE__, __LINE__);
            UnLock(m_lock);
            return C_OUT_OF_MEMORY;
        }

        UnLock(m_lock);

        //3.Wait receive thread send data and notify.
        ret = WaitForEvent(m_event, m_timeOut);

        if (m_status == running) { // TODO: FIXME: Still buggy
            Lock(m_lock);
            C_INT read = nToRead - m_receiveSize;
            memcpy((C_CHAR *)pBuffer, m_pReceiveAddr, read);
            *nRead = read;
            m_receiveSize = 0;
            m_receiveOffset = 0;
            FREE(m_pReceiveAddr);
            m_pReceiveAddr = NULL;
            UnLock(m_lock);
        }
        else { // may be deleted!
            if (m_lock != NULL) {
                Lock(m_lock);
            }
            m_receiveSize = 0;
            FREE(m_pReceiveAddr);
            m_pReceiveAddr = NULL;
            if (m_lock != NULL) {
                UnLock(m_lock);
            }
            return C_CLOSED;
        }

        if (0 != ret) {
            COMMU_PRINTF("Err: Read abort! file%s. line:%d.\n",
                __FILE__, __LINE__);
            if (C_FAIL == ret) {
                return C_CLOSED;
            }
            return ret;
        }
        ret = m_lasterror;
    }
    return ret;
}

C_INT CommunicationObject::Write(C_VOID *pBuffer,
    C_UINT NumberOfBytesToWrite,
    C_UINT *pNumberOfBytesWritten)
{
    C_INT ret = 0;
    if ((NULL == pBuffer) || (0 == NumberOfBytesToWrite) || (NULL == pNumberOfBytesWritten)) {
        return C_INVALID_ARGUMENT;
    }

    if (stopping == m_status) {
        COMMU_PRINTF("{ERR}: Channel closed.\r\n");
        return C_CLOSED;
    }

    if (pNumberOfBytesWritten != NULL) {
        *pNumberOfBytesWritten = 0;
    }
    int dataSize = NumberOfBytesToWrite;
    int packageCount = (dataSize + ChannelPackage::MAX_DATA_SIZE - 1) / ChannelPackage::MAX_DATA_SIZE;
    for (int i = 0; i < packageCount; i++) {
        int sendingSize = ChannelPackage::MAX_DATA_SIZE;
        if (i == packageCount - 1) {
            sendingSize = dataSize % ChannelPackage::MAX_DATA_SIZE;
            // last package size is 1~MAX_DATA_SIZE, NOT 0~(MAX_DATA_SIZE-1),
            // if dataSize is N*2K, last package will be dropped
            // fix it here
            if (sendingSize == 0) {
                sendingSize = ChannelPackage::MAX_DATA_SIZE;
            }
        }
        ChannelData cd(m_channelNumber, (char*) pBuffer + (i * ChannelPackage::MAX_DATA_SIZE), sendingSize);
        ret = SendChannelData(&cd);
        if (ret != 0) {
            return (0 == i ? ret : 0); // not totally
        }
        *pNumberOfBytesWritten += sendingSize;
    }

    return ret;
}

C_INT CommunicationObject::Init(PROPERTY property, C_UINT channelNumber)
{
    if (0 != channelNumber) {
        if (ChannelPackage::MAX_DATA_SIZE < channelNumber) {
            return C_INVALID_ARGUMENT;
        }
        m_channelNumber = channelNumber;
        Lock(m_sGenerateChannelNumberLock);
        if (channelNumber > m_sGenerateChannelNumber) {
            m_sGenerateChannelNumber = channelNumber;
        }
        UnLock(m_sGenerateChannelNumberLock);
    }
    else {
        Lock(m_sGenerateChannelNumberLock);
        if (m_sGenerateChannelNumber < MAX_CHANNEL_NUMBER) {
            m_channelNumber = ++m_sGenerateChannelNumber;
            UnLock(m_sGenerateChannelNumberLock);
        }
        else {
            UnLock(m_sGenerateChannelNumberLock);
            COMMU_PRINTF("Err: generate too much channel number! file%s. line:%d.\n",
                    __FILE__, __LINE__);
            return C_FAIL;
        }
    }

    m_event = NULL;

    m_property = property;
    m_status = connecting;
    m_lasterror = 0;
    m_timeOut = INFINITE;
    m_pReceiveStorageAddr = NULL;
    m_receiveStorageSize = 0;

    m_pReceiveAddr = NULL;
    m_receiveOffset = 0;
    m_receiveSize = 0;
    m_lock = InitLock();
    assert(m_lock);
    ListInit(&m_entry);
    return 0;
}

C_INT CommunicationObject::Stop()
{
    m_status = stopping;
    if (m_event != NULL) {
        EventSet(m_event);
    }

    m_channelNumber = 0x10000;
    m_lasterror = 0;
    if (NULL != m_event) {
        DeleEvent(m_event);
    }
    m_event = NULL;

    if (m_lock == NULL) return 0;

    Lock(m_lock);
    if (NULL != m_pReceiveStorageAddr) {
        FREE(m_pReceiveStorageAddr);
    }
    m_pReceiveStorageAddr = NULL;
    m_receiveStorageSize = 0;

    m_receiveSize = 0;
    UnLock(m_lock);

    DeleLock(m_lock);
    m_lock = NULL;
    ListRemove(&m_entry);
    return 0;
}

C_INT CommunicationObject::SendCreateOkReply()
{
    C_INT ret = 0;
    ChannelCommand cc(m_channelNumber, (C_INT)Reply, 0);
    ret = SendChannelData(&cc);
    if (C_FAILED(ret)) {
        COMMU_PRINTF("Err: SendCreateOKReply failed! file%s. line:%d. channel_number:%d\n",
                __FILE__, __LINE__, m_channelNumber);
    }

    return ret;
}

C_INT CommunicationObject::WaitCreateOkReply(C_INT *pCreateSucceed)
{
    C_INT ret = 0;
    *pCreateSucceed = 0;

    if (m_event == NULL) { // may be interrupted and stopped
        return C_CLOSED;
    }
    ret = WaitForEvent(m_event, INFINITE);
    if (C_FAILED(ret)) {
        if (C_TIME_OUT != ret)
            COMMU_PRINTF("Err: CommpoundRead failed! file%s. line:%d. channel_number:%d\n",
                    __FILE__, __LINE__, m_channelNumber);
        return ret;
    }

    EventReset(m_event);
    return ret;
}

C_INT CommunicationObject::Connect(C_INT* pCreateSucceed)
{
    C_INT ret = 0;
    C_INT CreateSucceed = 0;//FALSE
    C_INT count = 3;

    m_status = connecting;

    m_event = EventCreate(FALSE);
    if (NULL == m_event) {
        COMMU_PRINTF("ERR: Create Event failed! file%s. line:%d. channel_number:%d.\n",
                __FILE__, __LINE__, m_channelNumber);
        return C_FAIL;
    }

    Lock(CommunicationObject::m_sInitListLock);
    ListAddHead(&m_sInitList, &m_entry);
    UnLock(CommunicationObject::m_sInitListLock);

    while (count > 0) {
        if (0 != m_receiveStorageSize) {//The reply info may be lost or late, Has already receive data here.
            assert(m_pReceiveStorageAddr);
            CreateSucceed = 1;
            break;
        }
        ChannelCommand cc(m_channelNumber, ChannelCommand::TYPE_CREATE, (C_INT)m_property);
        ret = SendChannelData(&cc);
        if (C_FAILED(ret)) {
            if ((C_TIME_OUT == ret)) {
                continue;
            }
            COMMU_PRINTF("Err: SendCreateCommand failed! file%s. line:%d. channel_number:%d\n",
                    __FILE__, __LINE__, m_channelNumber);
            Lock(CommunicationObject::m_sInitListLock);
            ListRemove(&m_entry);
            UnLock(CommunicationObject::m_sInitListLock);
            return ret;
        }

        ret = WaitCreateOkReply(&CreateSucceed);
        if (C_FAILED(ret)) {
            if ((C_TIME_OUT == ret)) {
                continue;
            } // else C_CLOSED!
            COMMU_PRINTF("Err: WaitCreateOKReply failed! file%s. line:%d. channel_number:%d\n",
                    __FILE__, __LINE__, m_channelNumber);
            /*
             * Already closed in other thread!
             * As list has been re-initialized already,
             * there is no need to remove m_entry from list.
            Lock(CommunicationObject::m_sInitListLock);
            ListRemove(&m_entry);
            UnLock(CommunicationObject::m_sInitListLock);
            */
            return ret;
        }
        else {
            break;
        }
        count--;
    }

    if (NULL != pCreateSucceed) {
        *pCreateSucceed = CreateSucceed;
    }

    if (0 < count) {
        Lock(CommunicationObject::m_sInitListLock);
        ListRemove(&m_entry);
        UnLock(CommunicationObject::m_sInitListLock);
        Lock(CommunicationObject::m_sRunnListLock);
        ListAddHead(&m_sRunningList, &m_entry);
        UnLock(CommunicationObject::m_sRunnListLock);
        m_status = running;
    }

    return ret;
}

C_INT CommunicationObject::BeConnected()
{
    C_INT ret = 0;

    m_status = connecting;
    m_event = EventCreate(FALSE);
    if (NULL == m_event) {
        COMMU_PRINTF("Err: Create Event failed! file%s. line:%d. channel_number:%d.\n",
                __FILE__, __LINE__, m_channelNumber);
        return C_FAIL;
    }

    Lock(CommunicationObject::m_sRunnListLock);
    ListAddHead(&m_sRunningList, &m_entry);
    UnLock(CommunicationObject::m_sRunnListLock);

    ret = SendCreateOkReply();
    if (C_FAILED(ret)) {
        Lock(CommunicationObject::m_sRunnListLock);
        ListRemove(&m_entry);
        UnLock(CommunicationObject::m_sRunnListLock);
        COMMU_PRINTF("Err: SendCreateOKReply failed! file%s. line:%d. channel_number:%d\n",
                __FILE__, __LINE__, m_channelNumber);
    }

    return ret;
}

//user create a new channel: setUpFlag == 1
//user get an already channel: setUpFlag == 0
//receive thread create a new channel: channel number == setUpFlag>>16 and createFlag&&0xff == 1
C_INT CommunicationObject::CreateCommuObj(PROPERTY property,
    C_INT setUpFlag,
    CommunicationObject **ppObj)
{
    CommunicationObject *pObj = NULL;
    CommunicationObject *pconnectObj = NULL;
    list *pNode = NULL;
    C_INT ret = 0;
    C_INT channelNumber = setUpFlag>>16;
    if (0 != setUpFlag) {
        pObj = new CommunicationObject();
        if (NULL == pObj) {
            COMMU_PRINTF("Err: CreateCommuObj failed! file%s. line:%d.\n",
                    __FILE__, __LINE__);
            return C_OUT_OF_MEMORY;
        }
    }
    else {
        //match a connecting pObj
        Lock(CommunicationObject::m_sRunnListLock);
        pNode = NextNode(&m_sRunningList);
        while (pNode != &m_sRunningList) {
            pconnectObj = GET_LIST_OBJ(pNode, CommunicationObject, m_entry);
            if ((connecting == pconnectObj->m_status) && (property == pconnectObj->m_property)) {
                pconnectObj->m_status = running;
                *ppObj = pconnectObj;
                UnLock(CommunicationObject::m_sRunnListLock);
                return 0;
            }
            pNode = NextNode(pNode);
        }
        UnLock(CommunicationObject::m_sRunnListLock);
        return C_NO_OBJECT;
    }

    ret = pObj->Init(property, channelNumber);
    if (C_FAILED(ret)) {
        COMMU_PRINTF("Err: Init failed! file%s. line:%d.\n",
                __FILE__, __LINE__);
        goto EXIT;
    }

    if (0 != channelNumber) {
        ret = pObj->BeConnected();
        if (C_FAILED(ret)) {
            COMMU_PRINTF("Err: BeConnect failed! file%s. line:%d.\n",
                    __FILE__, __LINE__);
            goto EXIT;
        }
    }
    else {
        ret = pObj->Connect(NULL);
        if (C_FAILED(ret)) {
            COMMU_PRINTF("Err: Connect failed! file%s. line:%d.\n",
                    __FILE__, __LINE__);
            goto EXIT;
        }
    }

    if (NULL != ppObj) {
        *ppObj = pObj;
    }
    return 0;

EXIT:
    DeleCommuObj(pObj->m_channelNumber, TRUE);
    delete pObj;
    return ret;
}

CommunicationObject *FindObjByChannelNumber(list *pList, C_UINT channelNumber, C_LOCK lock);

C_VOID CommunicationObject::DeleCommuObj(C_UINT channelNumber, C_BOOL bBeDeleteFlag)
{
    CommunicationObject *pObj = NULL;
    pObj = FindObjByChannelNumber(&CommunicationObject::m_sRunningList,
                                 channelNumber,
                                 CommunicationObject::m_sRunnListLock);
    if (NULL == pObj) {
        return;
    }

    if (bBeDeleteFlag) {
        pObj->m_status = stopping;
        EventSet(pObj->m_event);
    }
    else {
        if (stopping != pObj->m_status) {
            ChannelCommand cc(channelNumber, ChannelCommand::TYPE_DELETE);
            SendChannelData(&cc);
        }
        Lock(CommunicationObject::m_sRunnListLock);
        pObj->Stop(); // ::Stop method has ListRemove(&m_entry); inside
        UnLock(CommunicationObject::m_sRunnListLock);
        //delete pObj;
    }
}

extern C_INT ReceiveLoop(C_VOID *arg);

C_VOID ReleaseChannelResources()
{
    if (CommunicationObject::m_sThreadHandle != NULL) {
        TerminateChannelThread(CommunicationObject::m_sThreadHandle);
        CommunicationObject::m_sThreadHandle = NULL;
    }
    if (CommunicationObject::m_sInitListLock != NULL) {
        DeleLock(CommunicationObject::m_sInitListLock);
        CommunicationObject::m_sInitListLock = NULL;
    }
    if (CommunicationObject::m_sRunnListLock != NULL) {
        DeleLock(CommunicationObject::m_sRunnListLock);
        CommunicationObject::m_sRunnListLock = NULL;
    }
    if (CommunicationObject::m_sGenerateChannelNumberLock != NULL) {
        DeleLock(CommunicationObject::m_sGenerateChannelNumberLock);
        CommunicationObject::m_sGenerateChannelNumberLock = NULL;
    }
}

C_INT CreateCompoundChannel()
{
    C_INT ret = 0;
    CommunicationObject::m_sGenerateChannelNumber = 0;

    CommunicationObject::m_sInitListLock = InitLock();
    if (CommunicationObject::m_sInitListLock == NULL) {
        COMMU_PRINTF("ERR: Init Lock fail! file%s. line:%d.\n",
                __FILE__, __LINE__);
        ReleaseChannelResources();
        return C_OUT_OF_MEMORY;
    }

    CommunicationObject::m_sRunnListLock = InitLock();
    if (CommunicationObject::m_sRunnListLock == NULL) {
        COMMU_PRINTF("ERR: Init Lock fail! file%s. line:%d.\n",
                __FILE__, __LINE__);
        ReleaseChannelResources();
        return C_OUT_OF_MEMORY;
    }

    CommunicationObject::m_sGenerateChannelNumberLock = InitLock();
    if (CommunicationObject::m_sGenerateChannelNumberLock == NULL) {
        COMMU_PRINTF("ERR: Init Lock fail! file%s. line:%d.\n",
                __FILE__, __LINE__);
        ReleaseChannelResources();
        return C_OUT_OF_MEMORY;
    }

    ListInit(&CommunicationObject::m_sInitList);
    ListInit(&CommunicationObject::m_sRunningList);
    ret = CreatePkgManager();
    if (0 != ret) {
        COMMU_PRINTF("ERR: CreatePkgManager fail! file%s. line:%d.\n",
                __FILE__, __LINE__);
        ReleaseChannelResources();
        return ret;
    }

    if (CommunicationObject::m_sThreadHandle == NULL) {
        ret = CreateChannelThread(ReceiveLoop, NULL, &CommunicationObject::m_sThreadHandle);
        if (0 != ret) {
            COMMU_PRINTF("ERR: CreateChannelThread (Receive) fail! file%s. line:%d.\n",
                    __FILE__, __LINE__);
            ReleaseChannelResources();
            return ret;
        }
    }

    ChannelCommand cc(ChannelCommand::TYPE_RESET);
    SendOutChannelData(&cc, false);

    return ret;
}

C_VOID DeleListCommuObj(list *pList, C_LOCK lock)
{
    assert(NULL != pList);

    CommunicationObject *pObj = NULL;
    list *pNode;
    Lock(lock);
    pNode = NextNode(pList);
    while (pNode != pList) {
        pObj = GET_LIST_OBJ(pNode, CommunicationObject, m_entry);
        pNode = NextNode(pNode);
        pObj->Stop();
//        delete pObj;
//        pObj = NULL;
    }
    ListInit(pList);
    UnLock(lock);
}

C_VOID  ResetCompoundChannel()
{
    DeleListCommuObj(&CommunicationObject::m_sInitList,
                CommunicationObject::m_sInitListLock);
    DeleListCommuObj(&CommunicationObject::m_sRunningList,
                CommunicationObject::m_sRunnListLock);
}

C_VOID  DeleteCompoundChannel()
{
    if (CommunicationObject::m_sThreadHandle != NULL) {
        // compound channel is still working
        ChannelCommand cc(ChannelCommand::TYPE_RESET);
        SendOutChannelData(&cc, false);
    }

    ResetCompoundChannel();

    ReleaseChannelResources();

    DelPkgManager();
}
